package org.example;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.*;

// 自制的mapreduce框架主程序
public class myhadoop {

    // 将原有的 main 方法改为 runMapReduce 方法
    public void runMapReduce(String[] args){
        Mapper mapobj = new Mapper();
        Reducer reducer = new Reducer();

        // map将键值对输出到 contextmap
        Context contextmap = new Context();

        // reduce将结果输出到 contextreduce
        Context contextreduce = new Context();

        // 将输入文件的每一行 line 输入到 map 函数, map 操作
        String paths = "org/example/wiki";
        String path1 = paths;
        String path2 = "";
        for(int i = 1; i <= 5; i++) {
            System.out.println("This is the " + i + "th job!");
            System.out.println("path1:" + path1);
            System.out.println("path2:" + path2);
            try{
                List<String> lines = Files.readAllLines(Paths.get(path1 + ".txt"), StandardCharsets.UTF_8);
                for(String line : lines){
                    mapobj.map(line, contextmap);
                }
            }catch (IOException e){
                e.printStackTrace();
            }
            // SHUFFLE
            // 排序
            ArrayList<KVPair> mapout = contextmap.read();
            Collections.sort(mapout);
            // 归并去重
            ArrayList<String> All_key = new ArrayList<>();
            for(KVPair shuffle_in : mapout){
                All_key.add(shuffle_in.read_key());
            }
            Set<String> shuffleout_keys = new HashSet<>(All_key);
            HashMap<String, ArrayList<String>> shuffle_result = new HashMap<>();
            for(String shuffleout_key : shuffleout_keys){
                ArrayList<String> shuffleout_value = new ArrayList<>();
                for(KVPair shuffle_in : mapout){
                    if(shuffleout_key.equals(shuffle_in.read_key())){
                        shuffleout_value.add(shuffle_in.read_value());
                    }
                }
                shuffle_result.put(shuffleout_key, shuffleout_value);
            }
            // reduce
            path2 = paths + i;
            for(String reducekey_in : shuffleout_keys){
                reducer.reduce(reducekey_in, shuffle_result.get(reducekey_in), contextreduce);
            }
            ArrayList<KVPair> reduce_outs = contextreduce.read();
            ArrayList<String> reduce_result = new ArrayList<>();
            for (KVPair reduce_out : reduce_outs) {
                reduce_result.add(reduce_out.read_key() + "\t" + reduce_out.read_value());
            }
            try{
                Files.write(Paths.get(path2 + ".txt"), reduce_result, StandardCharsets.UTF_8);
            }catch(IOException e){
                e.printStackTrace();
            }
            path1 = path2;
            contextmap.Arr_clear();
            shuffle_result.clear();
            contextreduce.Arr_clear();
        }
    }


}
